package org.springframework.cloud.gateway.filter.ratelimit;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.tuple.Tuple;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import static org.springframework.tuple.TupleBuilder.tuple;

/**
 * See https://stripe.com/blog/rate-limiters and
 * https://gist.github.com/ptarjan/e38f45f2dfe601419ca3af937fff574d#file-1-check_request_rate_limiter-rb-L11-L34
 *
 * @author Spencer Gibb
 */
public class RedisRateLimiter implements RateLimiter {
	public static final String REPLENISH_RATE_KEY = "replenishRate";
	public static final String BURST_CAPACITY_KEY = "burstCapacity";

	private Log log = LogFactory.getLog(getClass());

	private final ReactiveRedisTemplate<String, String> redisTemplate;
	private final RedisScript<List<Long>> script;

	public RedisRateLimiter(ReactiveRedisTemplate<String, String> redisTemplate, RedisScript<List<Long>> script) {
		this.redisTemplate = redisTemplate;
		this.script = script;
	}

	public static Tuple args(int replenishRate, int burstCapacity) {
		return tuple().of(REPLENISH_RATE_KEY, replenishRate, BURST_CAPACITY_KEY, burstCapacity);
	}

	/**
	 * This uses a basic token bucket algorithm and relies on the fact that Redis scripts
	 * execute atomically. No other operations can run between fetching the count and
	 * writing the new count.
	 * @param id 令牌桶编号，一个令牌桶编号对应令牌桶，在本文场景中为请求限流键
	 * @param args
	 * @return
	 */
	@Override
	@SuppressWarnings("unchecked")
	public Mono<Response> isAllowed(String id, Tuple args) {
		// How many requests per second do you want a user to be allowed to do?
		// 获得replenishRate，令牌桶填充平均速率，单位：秒
		int replenishRate = args.getInt(REPLENISH_RATE_KEY);

		// How much bursting do you want to allow?
		// 获得burstCapacity令牌桶上限
		int burstCapacity;
		if (args.hasFieldName(BURST_CAPACITY_KEY)) {
			burstCapacity = args.getInt(BURST_CAPACITY_KEY);
		} else {
			burstCapacity = 0;
		}

		try {
			// Make a unique key per user.
			// 获得令牌桶前缀，request_rate_limiter.${id}
			String prefix = "request_rate_limiter." + id;

			// You need two Redis keys for Token Bucket.
			// 获得令牌桶键数组：
			// request_rate_limiter.${id}.tokens：request_rate_limiter.${id}（令牌桶编号，一个令牌桶编号对应令牌桶），通过这个可以获取令牌桶剩余令牌数
			// request_rate_limiter.${id}.timestamp：request_rate_limiter.${id}（令牌桶编号，一个令牌桶编号对应令牌桶），通过这个可以令牌桶最后填充令牌时间，单位：秒
			List<String> keys = Arrays.asList(prefix + ".tokens", prefix + ".timestamp");

			// The arguments to the LUA script. time() returns unixtime in seconds.
			// 获得Lua脚本参数：
			// 第一个参数：replenishRate
			// 第二个参数：burstCapacity
			// 第三个参数：得到从1970-01-01 00:00:00开始的秒数。为什么在Java代码里获取，而不使用Lua在Reids里获取？
			// 因为Redis的限制（Lua中有写操作不能使用带随机性质的读操作，如TIME）不能在Redis Lua中使用TIME获取时间戳，因此只好从应用获取然后传入，在某些极端情况下（机器时钟不准的情况下），限流会存在一些小问题
			// 第四个参数：消耗令牌数量，默认1
			List<String> scriptArgs = Arrays.asList(replenishRate + "", burstCapacity + "",
					Instant.now().getEpochSecond() + "", "1");
			// allowed, tokens_left = redis.eval(SCRIPT, keys, args)
			// 调用ReactiveRedisTemplate#execute(RedisScript<T>, List<K>, List<?>)方法，执行Redis Lua脚本，获取令牌。返回结果为 [是否获取令牌成功，剩余令牌数]，其中，1代表获取令牌成功，0代表令牌获取失败
			Flux<List<Long>> flux = this.redisTemplate.execute(this.script, keys, scriptArgs);
					// .log("redisratelimiter", Level.FINER);
			return flux
                    // Throwable => Flux.just(Arrays.asList(1L, -1L)) 。
					// 当Redis Lua脚本过程中发生异常，忽略异常，返回Flux.just(Arrays.asList(1L, -1L))，即认为获取令牌成功。为什么？在Redis发生故障时，我们不希望限流器对Reids是强依赖，并且Redis发生故障的概率本身就很低
                    .onErrorResume(throwable -> Flux.just(Arrays.asList(1L, -1L)))
                    // Flux<List<Long>> => Mono<List<Long>>
					// 调用Flux#reduce(A, BiFunction<A, ? super T, A>)方法，将Flux<List<Long>>转换成Mono<List<Long>>。因为ReactiveRedisTemplate#execute(RedisScript<T>, List<K>, List<?>)的执行结果为Flux（多次），实际在当前场景里，执行Redis Lua脚本只会返回一次数组，所以转换成Mono（一次）
                    .reduce(new ArrayList<Long>(), (longs, l) -> {
						longs.addAll(l);
						return longs;
					})
                    // Mono<List<Long>> => Mono<Response>
                    // 调用Mono#map(Function<? super T, ? extends R>)方法，将Mono<List<Long>> => Mono<Response>
                    .map(results -> {
						boolean allowed = results.get(0) == 1L;
						Long tokensLeft = results.get(1);

						Response response = new Response(allowed, tokensLeft);

						if (log.isDebugEnabled()) {
							log.debug("response: " + response);
						}
						return response;
					})
                    ;
		}
		catch (Exception e) {
			/*
			 * We don't want a hard dependency on Redis to allow traffic. Make sure to set
			 * an alert so you know if this is happening too much. Stripe's observed
			 * failure rate is 0.01%.
			 */
			// 代码部分执行发生异常时，例如Redis挂了，返回Flux.just(Arrays.asList(1L, -1L))，即认为获取令牌成功
			log.error("Error determining if user allowed from redis", e);
		}
		return Mono.just(new Response(true, -1));
	}
}
